package com.kai.mqtt.protocol;

import com.kai.mqtt.auth.service.IMqttAclService;
import com.kai.mqtt.constant.AttributeKeyConstant;
import com.kai.mqtt.enums.AclActionEnum;
import com.kai.mqtt.message.RetainMessage;
import com.kai.mqtt.message.SubscribeMessage;
import com.kai.mqtt.message.service.IMessageIdService;
import com.kai.mqtt.message.service.IRetainMessageService;
import com.kai.mqtt.message.service.ISubscribeMessageService;
import com.kai.mqtt.session.Session;
import com.kai.mqtt.session.service.ISessionService;
import com.kai.mqtt.util.TopicUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * @author weijk
 * @since 2024/6/3
 */
@Slf4j
@Component
public class Subscribe extends AbstractProtocol {

    @Resource
    private ISubscribeMessageService subscribeMessageService;

    @Resource
    private IRetainMessageService retainMessageService;

    @Resource
    private IMessageIdService messageIdService;

    @Autowired
    private IMqttAclService mqttAclService;

    @Resource
    private ISessionService sessionService;

    @Override
    public void process(Channel channel, MqttMessage message) {
        MqttSubscribeMessage msg = (MqttSubscribeMessage) message;
        List<MqttTopicSubscription> topicSubscriptions = msg.payload().topicSubscriptions();
        String clientId = channel.attr(AttributeKeyConstant.CLIENT_ID).get();
        Session session = sessionService.get(clientId);

        List<Integer> subAckList = new ArrayList<>();
        for (MqttTopicSubscription topicSubscription : topicSubscriptions) {
            String topicFilter = topicSubscription.topicFilter();

            if (!mqttAclService.checkValid(session.getUsername(), clientId, AclActionEnum.SUBSCRIBE, topicFilter)) {
                subAckList.add(MqttQoS.FAILURE.value());
            } else {
                MqttQoS mqttQoS = topicSubscription.qualityOfService();
                SubscribeMessage subscribeMessage = new SubscribeMessage()
                        .setClientId(clientId)
                        .setTopic(topicFilter)
                        .setQos(mqttQoS.value());
                subscribeMessageService.put(topicFilter, subscribeMessage);
                subAckList.add(mqttQoS.value());
                log.info("receive - subscribe, clientId: {}, topicFilter: {}, qos: {}", clientId, topicFilter, mqttQoS.value());
            }
        }
        MqttSubAckMessage subAckMessage = (MqttSubAckMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                MqttMessageIdVariableHeader.from(msg.variableHeader().messageId()),
                new MqttSubAckPayload(subAckList));
        log.debug("send - subAck, clientId: {}, messageId: {}", clientId, msg.variableHeader().messageId());
        channel.writeAndFlush(subAckMessage);
        // 发布保留消息
        topicSubscriptions.forEach(topicSubscription -> {
            String topicFilter = topicSubscription.topicFilter();
            MqttQoS mqttQoS = topicSubscription.qualityOfService();
            if (!TopicUtil.isShareTopic(topicFilter)) {
                sendRetainMessage(channel, topicFilter, mqttQoS);
            }
        });
    }

    private void sendRetainMessage(Channel channel, String topicFilter, MqttQoS mqttQoS) {
        List<RetainMessage> retainMessages = retainMessageService.search(topicFilter);
        String clientId = channel.attr(AttributeKeyConstant.CLIENT_ID).get();
        retainMessages.forEach(retainMessage -> {
            MqttQoS respQoS = retainMessage.getQos() > mqttQoS.value() ? mqttQoS : MqttQoS.valueOf(retainMessage.getQos());
            if (Objects.equals(respQoS, MqttQoS.AT_MOST_ONCE)) {
                MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, false, respQoS, true, 0),
                        new MqttPublishVariableHeader(retainMessage.getTopic(), 0), Unpooled.buffer().writeBytes(retainMessage.getMessageBytes()));
                log.debug("send - publish, clientId: {}, topic: {}, qos: {}, content: {}", clientId, retainMessage.getTopic(), respQoS.value(), new String(retainMessage.getMessageBytes()));
                channel.writeAndFlush(publishMessage);
            }
            if (Objects.equals(respQoS, MqttQoS.AT_LEAST_ONCE) || Objects.equals(respQoS, MqttQoS.EXACTLY_ONCE)) {
                Integer messageId = messageIdService.get(clientId);
                MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, false, respQoS, true, 0),
                        new MqttPublishVariableHeader(retainMessage.getTopic(), messageId), Unpooled.buffer().writeBytes(retainMessage.getMessageBytes()));
                log.debug("send - publish, clientId: {}, topic: {}, qos: {}, messageId: {}, content: {}", clientId, retainMessage.getTopic(), respQoS.value(), messageId, new String(retainMessage.getMessageBytes()));
                channel.writeAndFlush(publishMessage);
            }
        });
    }
}
